package com.iot.service.mqtt.handler;

import com.iot.service.mqtt.model.request.TlMqttPublishReq;
import com.iot.service.mqtt.retry.RetryStrategy;
import com.iot.service.mqtt.retry.TlRetryMessage;
import com.iot.service.mqtt.service.message.IMessage;
import com.iot.service.mqtt.service.pub.IPublishStoreService;
import com.iot.service.mqtt.service.retain.IRetainService;
import com.iot.service.mqtt.service.session.ISessionService;
import com.iot.service.mqtt.service.subscripe.ISubscriptionService;
import com.iot.service.mqtt.utils.ChannelManager;
import io.netty.handler.codec.mqtt.MqttQoS;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Service;

/**
 * @Author: hszhou
 * @Date: 2025/5/8 13:45
 * @Description: 转发消息给客户端
 */
@Service
@RequiredArgsConstructor
@Slf4j
public class TlForwardMessageService {

    private final ISubscriptionService subscriptionService;

    private final IRetainService retainMsgService;

    private final IPublishStoreService publishStoreService;

    private final ISessionService sessionService;

    private final IMessage messageIdService;

    private final ChannelManager channelManager;

    private final RetryStrategy retryStrategy;


    /**
     * @description: 用于转发消息给订阅topic的客户端
     * @author: hszhou
     * @datetime: 2025-05-08 13:48:43
     * @param:
     * @param: topic
     * @param: req
     * @return: void
     **/
    public void forwardMessage(String topic, MqttQoS qoS,String content) {

        //找到所有的客户端
        subscriptionService.find(topic)
                .parallel()
                .subscribe(client->{
                    int subQos = client.getQos();
                    int realQos = Math.min(subQos, qoS.value());
                    MqttQoS mqttQoS = MqttQoS.valueOf(realQos);
                    String clientId = client.getClientId();
                    TlMqttPublishReq publishReq;
                    if(mqttQoS == MqttQoS.AT_MOST_ONCE){
                        publishReq = TlMqttPublishReq.build(topic,mqttQoS, false, content,0L);
                    }else{
                        Long messageId = messageIdService.getMessageId();
                        publishReq = TlMqttPublishReq.build(topic,mqttQoS, false, content,messageId);
                    }

                    //找到了客户端对应的通道 将publishReq 消息发送过去
                    sessionService.find(clientId)
                            .subscribe(session->{
                                Long messageId = publishReq.getVariableHead().getMessageId();
                                log.debug("将消息【{}】转发给【{}】", messageId, clientId);
                                channelManager.writeAndFlush(clientId, publishReq);

                                //如果消息类型是1或者2 那么就需要保留消息 防止在订阅的客户端cleanSession为0的时候 需要重新接收
                                if(mqttQoS == MqttQoS.EXACTLY_ONCE || mqttQoS == MqttQoS.AT_LEAST_ONCE){

                                    //保存消息时为了 有些客户端的cleansession是0  虽然断开了 但是后续连接时也需要重新发送 消息
                                    publishStoreService.savePublishMessage(clientId,messageId,publishReq).subscribe(res->{
                                        log.debug("保存消息【{}】成功", messageId);
                                        //设置重发消息  防止客户端没有接收到 需要重新发送
                                        TlRetryMessage retryMessage = new TlRetryMessage();
                                        retryMessage.setMessageId(messageId);
                                        retryMessage.setMessage(publishReq);
                                        publishReq.getFixedHead().setDup(true);
                                        retryStrategy.retry( retryMessage,clientId);
                                    });
                                }
                            });

                });

    }
}
